package com.zhang.flink.exercise;

import com.zhang.flink.bean.AdCount;
import com.zhang.flink.bean.AdsClickLog;
import org.apache.flink.api.common.eventtime.SerializableTimestampAssigner;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.AggregateFunction;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.common.state.ValueState;
import org.apache.flink.api.common.state.ValueStateDescriptor;
import org.apache.flink.api.common.typeinfo.Types;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.KeyedProcessFunction;
import org.apache.flink.streaming.api.functions.windowing.ProcessWindowFunction;
import org.apache.flink.streaming.api.windowing.assigners.SlidingEventTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.util.Collector;
import org.apache.flink.util.OutputTag;


/**
 * @title: 每隔五分钟统计过去一个小时各省份广告点击量
 * 同时对同一个用户点击同一个广告超过100次拉入黑名单处理，后续不再进行总数统计
 * @author: zhangyifan
 * @date: 2022/12/8 14:10
 */
public class AdClickAnalysis {
    public static void main(String[] args) throws Exception {
        // 获取执行环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);

        // 获取数据源分配水位线和提取时间戳
        SingleOutputStreamOperator<AdsClickLog> adsClickLogDS = env
                .readTextFile("/Users/apple/project-git/flink-example/src/main/resources/AdClickLog.csv")
                .map(new MapFunction<String, AdsClickLog>() {
                    @Override
                    public AdsClickLog map(String value) throws Exception {
                        String[] fields = value.split(",");
                        return new AdsClickLog(
                                fields[0], fields[1], fields[2], fields[3],
                                Long.parseLong(fields[4])
                        );
                    }
                })
                .assignTimestampsAndWatermarks(
                        WatermarkStrategy.<AdsClickLog>forMonotonousTimestamps()
                                .withTimestampAssigner(new SerializableTimestampAssigner<AdsClickLog>() {
                                    @Override
                                    public long extractTimestamp(AdsClickLog element, long recordTimestamp) {
                                        return element.ts * 1000L;
                                    }
                                })
                );
        // 把有刷单行为的用户信息输出到黑名单(侧输出流中)并做过滤
        SingleOutputStreamOperator<AdsClickLog> warningDS = adsClickLogDS
                .keyBy(r -> r.userId + "_" + r.adId)
                .process(new KeyedProcessFunction<String, AdsClickLog, AdsClickLog>() {
                    private Long maxCountClick = 100L;

                    // 保存点击次数的状态
                    private ValueState<Long> clickCountState;
                    // 标识用户是否被拉入黑名单
                    private ValueState<Boolean> isBlackList;

                    @Override
                    public void open(Configuration parameters) throws Exception {
                        clickCountState = getRuntimeContext().getState(
                                new ValueStateDescriptor<Long>(
                                        "click-count-state", Types.LONG
                                )
                        );

                        isBlackList = getRuntimeContext().getState(
                                new ValueStateDescriptor<Boolean>(
                                        "is-black-list", Types.BOOLEAN
                                )
                        );
                    }

                    @Override
                    public void processElement(AdsClickLog value, KeyedProcessFunction<String, AdsClickLog, AdsClickLog>.Context ctx, Collector<AdsClickLog> out) throws Exception {
                        Long count = clickCountState.value();
                        if (count == null) {
                            clickCountState.update(0L);
                            long ts = (ctx.timerService().currentProcessingTime() / (1000 * 60 * 60 * 24) + 1) * (24 * 60 * 60 * 1000L) - 8 * 60 * 60 * 1000L;
                            ctx.timerService().registerProcessingTimeTimer(ts);
                        } else {
                            count = count + 1L;
                            clickCountState.update(count);
                            if (count >= maxCountClick) {
                                if (isBlackList.value() == null) {
                                    ctx.output(new OutputTag<String>("warning") {
                                    }, value.userId + "点击了" + value.adId + "广告达到" + maxCountClick + "次，存在恶意点击来入黑名单，报警！！！");
                                    //加入黑名单
                                    isBlackList.update(true);
                                }
                                return;
                            }
                        }
                        out.collect(value);
                    }
                });

        // 每隔五分钟统计过去一个小时各省份广告点击量
        SingleOutputStreamOperator<AdCount> resultDS = warningDS
                .keyBy(r -> r.province)
                .window(SlidingEventTimeWindows.of(Time.hours(1), Time.minutes(5)))
                .aggregate(
                        new AggregateFunction<AdsClickLog, Long, Long>() {
                            @Override
                            public Long createAccumulator() {
                                return 0L;
                            }

                            @Override
                            public Long add(AdsClickLog value, Long accumulator) {
                                return accumulator + 1L;
                            }

                            @Override
                            public Long getResult(Long accumulator) {
                                return accumulator;
                            }

                            @Override
                            public Long merge(Long a, Long b) {
                                return a + b;
                            }
                        },
                        new ProcessWindowFunction<Long, AdCount, String, TimeWindow>() {
                            @Override
                            public void process(String key, ProcessWindowFunction<Long, AdCount, String, TimeWindow>.Context context, Iterable<Long> elements, Collector<AdCount> out) throws Exception {
                                out.collect(new AdCount(
                                        key,
                                        elements.iterator().next(),
                                        context.window().getStart(),
                                        context.window().getEnd()
                                ));
                            }
                        }
                );

        // todo 提取侧输出流报警数据
        warningDS.getSideOutput(new OutputTag<String>("warning") {
        }).print("warning:");
        // todo 打印各省份点击情况
        resultDS.print();
        // 执行任务
        env.execute();
    }
}
